%% Created: 2012-4-5
%% Description: TODO: Add description to GossipNode
-module(gossip_node).

%%
%% Include files
%%

%%
%% Exported Functions
%%
-export([init/3]).

%%
%% API Functions
%%

%%
%% Entry point for gossip_node process, passed a unique ID in the network as the identification for each node.
%% Wait until "start" or "stop" message received. In the "start" message, the PID of the monitor process as well
%% as all the neighbors PIDs and this nodes initial configuration are included.
%%
%% Parameters:
%% 	ID: unique ID for this node
%%
init(ID, Vc, Vm) ->
	gossip:init_random(),
	put('CPU', 1000),
	put('MEM', 1000),
	put('Vc', Vc),
	put('Vm', Vm),
	receive
		{start, _MainPid, MonitorPid, Nodes, GossipData} ->
			Neighbours = [{NID, Pid} || {NID, Pid}  <- Nodes, NID/=ID],
			loop(MonitorPid, {ID, self()}, Neighbours, GossipData);
		{stop} ->
			ok
	end.

%%
%% Loop used after sending a gossip request to wait for an acknowledgement from the receiver. 
%%
%% Different types of received messages are handled:
%%	"ok" message: correct acknowledgement, update this nodes configuration and go back to the main loop.
%% 	"reject" message: rejected by the receiver, the sended request failed, go back to the main loop.
%%	REQUEST from other nodes: send reject, and continue to wait the response.
%%	"stop" message: shutdown process, return "ok".
%% 
%% In handling of the different types of messages, different stat information is sent to the gossip_monitor.
%% Format: 
%% 	{{ActiveSend, ActiveReceiveNormal, ActiveReceiveReject, ActiveReceiveTimeout}, {PassiveSendNormal, PassiveSendReject}}
%%
%% Parameters:
%% 	MonitorPid: PID for the gossip_monitor
%%	Node: {ID, PID} for the current node
%% 	ToNode: {ID, PID} for the receiver node
%%	Neighbours: Array of {ID, PID} for the neighbours
%% 	GossipData: Internal data for the current node, here means the CPU configuration
%%
wait_send_response(CPU, MEM, MonitorPid, Node, ToNode, Neighbours, GossipData) ->
	receive
		{response, ToNode, ok, GossipDataNew} ->
			% active receive normal {{0, 1, 0, 0}, {0, 0}}		
			%io:format("R(~p): response(ok), ~p (~p) . [~p=>~p]... ~n", [Node, ToNode, GossipDataFromResponse, GossipData, GossipDataNew]),
			MonitorPid ! {stat, element(1, Node), {{0, 1, 0, 0}, {0, 0}}},
			MonitorPid ! {report, element(1, Node), {GossipDataNew, {CPU, MEM}}},
			loop(MonitorPid, Node, Neighbours, GossipDataNew);
		
		{response, ToNode, reject} ->
			% active receive reject {{0, 0, 1, 0}, {0, 0}}
			%io:format("R(~p): response(rj), ~p ... ~n", [Node, ToNode]),
			MonitorPid ! {stat, element(1, Node), {{0, 0, 1, 0}, {0, 0}}},
			loop(MonitorPid, Node, Neighbours, GossipData);
	
		{request, FromNodeReject, _GossipDataFrom} ->
			%io:format("R(~p): request(rj), ~p... ~n", [Node, FromNodeReject]),
			% passive send reject  {{0, 0, 0, 0}, {0, 1}}
			element(2, FromNodeReject) ! {response, Node, reject},
			MonitorPid ! {stat, element(1, Node), {{0, 0, 0, 0}, {0, 1}}},
			wait_send_response(CPU, MEM, MonitorPid, Node, ToNode, Neighbours, GossipData);
		
		{stop} ->
			ok
	end.

%%
%% Main loop for the gossip_node, node is first waiting for a randomed period of time for the gossip request,
%% if received, this nodes configuration is updated and an acknowledgement is sent and then continue the loop;
%% if timeout, the node will randomly select a neighbour to send the gossip request, and then enter wait_send_response.
%% Stat information is also sent to gossip_monitor when there is any send ORreceive activities. The same format is used for stats.
%%
%% Different types of received messages are handled:
%%	REQUEST from other nodes: update this nodes configuration, send acknowledgement, and continue the loop.
%%	"stop" message: shutdown process, return "ok".
%% 
%% Parameters:
%% 	MonitorPid: PID for the gossip_monitor
%%	Node: {ID, PID} for the current node
%%	Neighbours: Array of {ID, PID} for the neighbours
%% 	GossipData: Internal data for the current node, here means the CPU configuration
%%
loop(MonitorPid, Node, Neighbours, GossipData) ->
	%set new MEM, new CPU,	
	CPUOld = get('CPU'),
	MEMOld = get('MEM'),
	Vc = get('Vc'),
	Vm = get('Vm'),
	CPU = gossip:random(CPUOld*(1-Vc), CPUOld*(1+Vc)),
	MEM = gossip:random(MEMOld*(1-Vm), MEMOld*(1+Vm)),
	Timeout = get_time_out(length(Neighbours)+1),
	receive
		{request, FromNode, GossipFrom} ->
			%received information has {CPU, MEM}; but response doens't
			{GossipDataNew, GossipDataFromNew} = share({GossipData, {CPU, MEM}}, GossipFrom),
			%io:format(">>>>~n~p~n>>~n~p~n=~n~p~n>>~n~p~n----~n", [{GossipData, {CPU, MEM}}, GossipFrom, GossipDataNew, GossipDataFromNew]),
			%io:format("R(~p): request(ok), ~p (~p). [~p=>~p]... ~n", [Node, FromNode, GossipDataFrom, GossipData, GossipDataNew]),			
			element(2, FromNode) ! {response, Node, ok, GossipDataFromNew},
			
			% passive send normal  {{0, 0, 0, 0}, {1, 0}}
			MonitorPid ! {stat, element(1, Node), {{0, 0, 0, 0}, {1, 0}}},
			MonitorPid ! {report, element(1, Node), {GossipDataNew, {CPU, MEM}}},
			loop(MonitorPid, Node, Neighbours, GossipDataNew);
		
		{stop} ->
			ok
	after
			Timeout ->
				% active send {{1, 0, 0, 0}, {0, 0}}
				ToNode = select_neighbour(Neighbours),
				%io:format("S(~p), ~p(~p)...~n", [Node, ToNode, GossipData]),
				element(2, ToNode) ! {request, Node, {GossipData, {CPU, MEM}}},		
				MonitorPid ! {stat, element(1, Node), {{1, 0, 0, 0}, {0, 0}}},
				wait_send_response(CPU, MEM, MonitorPid, Node, ToNode, Neighbours, GossipData)
	end.

share(GossipReceive, GossipSend)->
	{GossipDataReceive, {CPUR, _MEMR}} = GossipReceive,
	{GossipDataSend, {CPUS, _MEMS}} = GossipSend,
	{_IDR, TCPUR, _TMEMR, _PLANR} = GossipDataReceive,
	{_IDS, TCPUS, _TMEMS, _PLANS} = GossipDataSend,
	if
		TCPUR/CPUR >= TCPUS/CPUS ->
			share_ordered(GossipReceive, GossipSend);
		true ->
			{B, A} = share_ordered(GossipSend, GossipReceive),
			{A, B}
	end.

share_ordered(GossipL, GossipS)->
	{GossipDataL, {CPUL, _MEML}} = GossipL,
	{GossipDataS, {CPUS, MEMS}} = GossipS,
	{IDL, TCPUL, _TMEML, PLANL} = GossipDataL,
	{IDS, TCPUS, TMEMS, PLANS} = GossipDataS,
	DeltaCPU = (TCPUL * CPUS - TCPUS * CPUL) div (CPUL + CPUS),
	if
		DeltaCPU >0 ->
			{{PLANLC, PLANLD}, {PLANSC, PLANSD}} = split(PLANL, PLANS),
			if
				length(PLANLC) > 0 ->
					PLANLCNew = sort_unit_mem(PLANLC, -1),
					{PLANLCNewShared, PLANSCShared, DeltaCPUNew} = share_common(PLANLCNew, PLANSC, DeltaCPU),
					if
						DeltaCPUNew == 0 ->
							calc_plan(IDL, lists:append(PLANLCNewShared, PLANLD), IDS, lists:append(PLANSCShared, PLANSD));
						true ->
							share_ordered_next(PLANLCNewShared, PLANLD, PLANSCShared, PLANSD, DeltaCPUNew, MEMS-TMEMS, IDL, IDS)
					end;
				true ->
					share_ordered_next(PLANLC, PLANLD, PLANSC, PLANSD, DeltaCPU, MEMS-TMEMS, IDL, IDS)
			end;
		true ->
			{GossipDataL, GossipDataS}
	end.

share_ordered_next(PLANLC, PLANLD, PLANSC, PLANSD, DeltaCPU, MAXMEM, IDL, IDS) ->
	if
		length(PLANLD) >0 ->
			PLANLDNew = sort_unit_mem(PLANLD, 1),
			{PLANLDNewShared, PLANSDShared, _DeltaCPUNew} = share_different(PLANLDNew, PLANSD, DeltaCPU, MAXMEM),
			calc_plan(IDL, lists:append(PLANLC, PLANLDNewShared), IDS, lists:append(PLANSC, PLANSDShared));
		true ->
			calc_plan(IDL, lists:append(PLANLC, PLANLD), IDS, lists:append(PLANSC, PLANSD))
	end.

calc_plan(IDL, PLANL, IDS, PLANS)->
	{calc_plan(IDL, PLANL), calc_plan(IDS, PLANS)}.

calc_plan(IDL, PLANL)->
	CPUList = [CPU || {_CID, CPU, _MEM} <- PLANL],
	MEMList = [MEM || {_CID, _CPU, MEM} <- PLANL],
	TakenCPU = lists:sum(CPUList),
	TakenMEM = lists:sum(MEMList),
	{IDL, TakenCPU, TakenMEM, PLANL}.

share_different(PLANLD, PLANSD, DeltaCPU, MAXMEM)->
	PLANs = prepare_plans(PLANLD, [{[], 0, 0}], 1, DeltaCPU, MAXMEM),
	{CIDs, CPU, MEM} = lists:nth(1, PLANs),
	DeltaCPUNew = DeltaCPU - CPU,
	MAXMEMNew = MAXMEM - MEM,
	ListL = [{CID, {CPU, MEM}} || {CID, CPU, MEM} <- PLANLD],
	DictL = dict:from_list(ListL),
	{DictLNew, PLANLDNew, PLANSDNew} = share_different_exec(DictL, PLANSD, CIDs, 1),
	if
		DeltaCPUNew == 0 ->
			{PLANLDNew, PLANSDNew, 0};
		true ->			
			PLANLDNewFiltered = [{CID, CPU, MEM} || {CID, CPU, MEM} <- PLANLDNew, CPU >= DeltaCPUNew, MEM =< MAXMEMNew],
			if
				length(PLANLDNewFiltered) >0 ->
					PLANLDNewFilteredSorted = sort_mem(PLANLDNewFiltered, 1),
					{CIDF, CPUF, MEMF} = lists:nth(1, PLANLDNewFilteredSorted),
					DictLNewNew = dict:store(CIDF, {CPUF - DeltaCPUNew, MEMF}, DictLNew),
					PLANSDNewNew = lists:append(PLANSDNew, [{CIDF, DeltaCPUNew, MEMF}]),
					PLANLDNewNew = dict_to_list(DictLNewNew),
					{PLANLDNewNew, PLANSDNewNew, 0};
				true->
					{PLANLDNew, PLANSDNew, DeltaCPUNew}
			end
	end.

share_different_exec(DictL, PLANSD, CIDs, Index)->
	if
		Index > length(CIDs)->
			PLANLD = dict_to_list(DictL),
			{DictL, PLANLD, PLANSD};
		true->
			CID = lists:nth(Index, CIDs),
			{CPU, MEM} = dict:fetch(CID, DictL),
			PLANSDNew = lists:append(PLANSD, [{CID, CPU, MEM}]),
			DictLNew = dict:erase(CID, DictL),
			share_different_exec(DictLNew, PLANSDNew, CIDs, Index+1)
	end.

prepare_plans(PLANLD, Current, Index, DeltaCPU, MAXMEM)->
	if
		Index > length(PLANLD)->
			Current;
		true ->
			{CID, CPU, MEM} = lists:nth(Index, PLANLD),
			CurrentNewGenerated = [{lists:append(EIDs, [CID]), ECPU + CPU, EMEM + MEM} || {EIDs, ECPU, EMEM} <- Current, ECPU + CPU =< DeltaCPU, EMEM + MEM =< MAXMEM],
			CurrentNew = lists:merge(Current, CurrentNewGenerated),
			CurrentNewFiltered = filter(CurrentNew),
			prepare_plans(PLANLD, CurrentNewFiltered, Index+1, DeltaCPU, MAXMEM)
	end.

filter(PLAN)->
	if
		length(PLAN) ==0 ->
			PLAN;
		true ->
			PLANSorted = sort_cpu(PLAN, -1),
			First = lists:nth(1, PLANSorted),
			filter(PLANSorted, [First], 2)
	end.
	
filter(PLAN, PLANNew, Index)->
	if
		Index > length(PLAN)->
			PLANNew;
		true ->
			Latest = lists:nth(length(PLANNew), PLANNew),
			{_CIDsLatest, _CPULatest, MEMLatest} = Latest,
			Current = lists:nth(Index, PLAN),
			{_CIDs, _CPU, MEM} = Current,
			if
				MEM < MEMLatest ->
					filter(PLAN, lists:append(PLANNew, [Current]), Index+1);
				true ->
					filter(PLAN, PLANNew, Index+1)
			end
	end.
	

share_common(PLANLC, PLANSC, DeltaCPU)->
	ListS = [{CID, {CPU, MEM}} || {CID, CPU, MEM} <- PLANSC],
	DictS = dict:from_list(ListS),
	share_common(PLANLC, DictS, DeltaCPU, 1).
	
share_common(PLANLC, DictS, DeltaCPU, Index) ->
	if
		Index>length(PLANLC)->
			PLANSC = dict_to_list(DictS),
			{[], PLANSC, DeltaCPU};
		true->
			CurrentItem = lists:nth(Index, PLANLC),
			{CID, CPU, MEM} = CurrentItem,
			{TakenCPU, TakenMEM} = dict:fetch(CID, DictS),
			if
				DeltaCPU > CPU->
					DictSNew = dict:store(CID, {TakenCPU + CPU, TakenMEM}, DictS),
					share_common(PLANLC, DictSNew, DeltaCPU - CPU, Index + 1);
				true->
					DictSNew = dict:store(CID, {TakenCPU + DeltaCPU, TakenMEM}, DictS),
					PLANLC1 = lists:sublist(PLANLC, Index+1, length(PLANLC) - Index),
					if
						CPU - DeltaCPU > 0->
							PLANLCNew = lists:append(PLANLC1, [{CID, CPU - DeltaCPU, MEM}]);
						true ->
							PLANLCNew = PLANLC1
					end,
					PLANSC = dict_to_list(DictSNew),
					{PLANLCNew, PLANSC, 0}
			end
	end.
					
				
dict_to_list(Dict) ->
	DictDList = dict:to_list(Dict),
	lists:map(fun({K,V})->
			{CPU, MEM} = V,
			{K, CPU, MEM}
			end, DictDList).

split(PLANL, PLANS)->
	CIDsL = [CID || {CID, _CPU, _MEM}<-PLANL],
	CIDsS = [CID || {CID, _CPU, _MEM}<-PLANS],
	IDSetL = sets:from_list(CIDsL),
	IDSetS = sets:from_list(CIDsS),
	IDSetCommon = sets:intersection(IDSetL, IDSetS),
	PLANLC = [{CID, CPU, MEM} || {CID, CPU, MEM}<-PLANL, sets:is_element(CID, IDSetCommon) == true],
	PLANLD = [{CID, CPU, MEM} || {CID, CPU, MEM}<-PLANL, sets:is_element(CID, IDSetCommon) == false],
	PLANSC = [{CID, CPU, MEM} || {CID, CPU, MEM}<-PLANS, sets:is_element(CID, IDSetCommon) == true],
	PLANSD = [{CID, CPU, MEM} || {CID, CPU, MEM}<-PLANS, sets:is_element(CID, IDSetCommon) == false],
	{{PLANLC, PLANLD}, {PLANSC, PLANSD}}.

sort_unit_mem(PLAN, Direction) ->
	List = [{CID, {CPU, MEM}} || {CID, CPU, MEM} <- PLAN],
	Dict = dict:from_list(List),
	OrdList = [{{MEM/CPU*Direction, CID}, CID} || {CID, CPU, MEM} <- PLAN],
	OrdDict = orddict:from_list(OrdList),
	OrdListNew = orddict:to_list(OrdDict),
	lists:map(fun({_K,V})->
						{CPU, MEM} = dict:fetch(V, Dict),
						{V, CPU, MEM}
						end, OrdListNew).

sort_mem(PLAN, Direction) ->
	List = [{CID, {CPU, MEM}} || {CID, CPU, MEM} <- PLAN],
	Dict = dict:from_list(List),
	OrdList = [{{MEM*Direction, CID}, CID} || {CID, _CPU, MEM} <- PLAN],
	OrdDict = orddict:from_list(OrdList),
	OrdListNew = orddict:to_list(OrdDict),
	lists:map(fun({_K,V})->
						{CPU, MEM} = dict:fetch(V, Dict),
						{V, CPU, MEM}
						end, OrdListNew).

sort_cpu(PLAN, Direction) ->
	List = [{CID, {CPU, MEM}} || {CID, CPU, MEM} <- PLAN],
	Dict = dict:from_list(List),
	OrdList = [{{CPU*Direction, CID}, CID} || {CID, CPU, _MEM} <- PLAN],
	OrdDict = orddict:from_list(OrdList),
	OrdListNew = orddict:to_list(OrdDict),
	lists:map(fun({_K,V})->
						{CPU, MEM} = dict:fetch(V, Dict),
						{V, CPU, MEM}
						end, OrdListNew).
get_time_out(Size) ->
	MaxBound = 2000,
	MinBound = 200,
	Bound = erlang:min(MaxBound,erlang:max(MinBound, Size*10)),
	random:uniform(Bound).

select_neighbour(Neighbours) ->
	lists:nth(random:uniform(length(Neighbours)), Neighbours).
